package com.pig4cloud.pigx.contract.websocket.subscribe;

import com.alibaba.fastjson.JSONObject;
import com.pig4cloud.pigx.contract.entity.ContractKLine;
import com.pig4cloud.pigx.contract.utils.ContractUtils;
import com.pig4cloud.pigx.contract.vo.AbstractTopicVO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class Kline extends DataListAbstractTopic {

	protected String TOPICNAME = "Kline";

	private final ContractUtils utils;

	private final RedisTemplate<String,Object> redis;

	public Kline(ContractUtils utils, RedisTemplate redis) {
		this.utils = utils;
		this.redis = redis;
		init();
	}

	private void init() {
		wsPush();
	}

	private void wsPush() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor();
		service.scheduleAtFixedRate(() -> {
			try {
				// ----------该推送方案 仅限于推送信息与用户信息无关联的场景----------
				// ----------实现方式为 将所有session根据查询类型进行划分,大幅度减少查询消耗,可承受大面积用户同时访问----------
				// ----------在服务器允许的情况下 可以将每种数据类型都new一个线程去执行,可大幅度提升并发访问的推送效率----------
				// 数据分类容器
				Map<String, List<WebSocketSession>> map = new HashMap<>();
				// 用户原生请求容器
				Map<String, List<String>> paramsMap = new HashMap<>();
				// 统计所有订阅K线用户的订阅类型
				for (Map.Entry<WebSocketSession, AbstractTopicVO> entry : subList.entrySet()) {
					if (!entry.getKey().isOpen()) {
						subList.remove(entry.getKey());
						log.info("ws 移除已关闭的session");
						continue;
					}

					// 请求参数解析
					AbstractTopicVO param = entry.getValue();
					List<String> dataList = param.getDataList();
					if (dataList.size() < 4) {
						continue;
					}

					// k线类型
					String key = dataList.get(1) + "," + dataList.get(2);
					// 如果已有该类型,则在该k线类型中添加发送用户
					if (map.containsKey(key)) {
						map.get(key).add(entry.getKey());
					} else {
						List<WebSocketSession> sessionList = new ArrayList<>();
						sessionList.add(entry.getKey());
						map.put(key, sessionList);
						// 保存用户原生请求
						paramsMap.put(key, dataList);
					}
				}

				// 遍历所有需要发送的k线类型,将各类型数据分发给订阅的用户
				map.forEach((key, value) -> {
					String[] split = key.split(",");
					String symbolName = split[0];
					String lineType = split[1];
					ContractKLine kLine = utils.selectCurrentKLine(symbolName, lineType);
					if (kLine != null) {
						// 时间戳
						long l = kLine.getStartTime().toInstant(ZoneOffset.ofHours(8)).toEpochMilli();
						kLine.setTimeStamp(l);
						// 获取币对最新价
						String redisKey = symbolName+"_last_price";
						Object price  = redis.opsForValue().get(redisKey);
						if (price != null) {
							BigDecimal lastPrice =new  BigDecimal(price.toString());
							kLine.setClosePrice(lastPrice);
						}
						// 参数组装
						JSONObject object = this.getJsonObject(kLine, lineType);
						TextMessage message = getTextMessage(paramsMap.get(key), object);
						// 将数据发送给所有订阅该K线的用户
						value.forEach(session -> {
							try {
								session.sendMessage(message);
							} catch (IOException e) {
								log.info("push kline err:" + e.getMessage());
							}
						});
					}
				});
			} catch (Exception e) {
				e.printStackTrace();
				log.info("push kline err:" + e.getMessage());
			}
		}, 0, 200, TimeUnit.MILLISECONDS);
	}

	private JSONObject getJsonObject(ContractKLine kLine, String type) {
		JSONObject object = new JSONObject();
		object.put("e", "Kline");
		object.put("E", LocalDateTime.now().toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
		object.put("s", kLine.getSymbolName());
		object.put("k", new JSONObject() {{
			put("t", kLine.getStartTime().toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
			put("T", kLine.getEndTime().toInstant(ZoneOffset.ofHours(8)).toEpochMilli());
			put("s", kLine.getSymbolName());
			put("i", type);
			put("f", "");
			put("L", "");
			put("o", kLine.getOpenPrice());
			put("c", kLine.getClosePrice());
			put("h", kLine.getHighPrice());
			put("l", kLine.getLowPrice());
			put("v", kLine.getDealNum());
			put("n", "");
			put("x", false);
			// 涨跌额
			BigDecimal subtract = kLine.getClosePrice().subtract(kLine.getOpenPrice());
			if (subtract.compareTo(BigDecimal.ZERO) < 0) {
				put("q", subtract.negate());
			} else {
				put("q", subtract);
			}
			// 涨跌幅
			BigDecimal divide = subtract.multiply(BigDecimal.TEN).multiply(BigDecimal.TEN).divide(kLine.getOpenPrice(),2,BigDecimal.ROUND_HALF_UP);
			put("Q", divide);
			put("V", kLine.getDealPrice());
			put("B", "");
		}});
		return object;
	}
}
